/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.client.consumer.rebalance;

import java.util.ArrayList;
import java.util.List;
import org.apache.rocketmq.common.message.MessageQueue;

/**
 * Average Hashing queue algorithm
 */
public class AllocateMessageQueueAveragely extends AbstractAllocateMessageQueueStrategy {

    @Override
/**
 * 根据平均分配策略为当前消费者分配消息队列。
 *
 * @param consumerGroup 消费者组名称。
 * @param currentCID    当前消费者的客户端ID。
 * @param mqAll         所有消息队列的列表。
 * @param cidAll        所有消费者客户端ID的列表。
 * @return 返回分配给当前消费者的消息队列列表。
 */
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {
        // 初始化结果列表
        List<MessageQueue> result = new ArrayList<MessageQueue>();

        // 检查输入参数是否有效，如果无效则返回空列表
        if (!check(consumerGroup, currentCID, mqAll, cidAll)) {
            return result;
        }

        // 获取当前消费者在消费者列表中的索引
        int index = cidAll.indexOf(currentCID);

        // 计算消息队列总数对消费者总数的余数
        int mod = mqAll.size() % cidAll.size();

        // 计算每个消费者平均分配的消息队列数量
        int averageSize =
                mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size() + 1 : mqAll.size() / cidAll.size());

        // 计算当前消费者分配消息队列的起始索引
        // 当前这个 Consumer 从 MessageQueue 数组的哪个位置开始取
        int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;

        // 计算当前消费者分配消息队列的范围
        // 这个 Consumer 获取到了多少个 MessageQueue
        int range = Math.min(averageSize, mqAll.size() - startIndex);

        // 根据起始索引和范围，将消息队列分配给当前消费者
        for (int i = 0; i < range; i++) {
            result.add(mqAll.get((startIndex + i) % mqAll.size()));
        }

        // 返回分配结果
        return result;
    }


    @Override
    public String getName() {
        return "AVG";
    }
}
